package com.zdb.demo.flink.streaming;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 实现三个 Flink Stream Job，分别达到如下效果，并截图
 * 造成 HIGH 背压，并且导致 Checkpoint 失败
 * 引起 LOW 背压，并且 Checkpoint 耗时长
 * 无背压，并且 Checkpoint 正常
 */
public class H9 {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        streamEnv.enableCheckpointing(2000, CheckpointingMode.AT_LEAST_ONCE);
        streamEnv.setStateBackend(new MemoryStateBackend());
        backPressureHigh(streamEnv);
//        backPressureLow(streamEnv);
//        backPressureNo(streamEnv);
        streamEnv.execute();
    }

    private static void backPressureHigh(StreamExecutionEnvironment streamEnv) {
        streamEnv.fromCollection(IntStream.range(0, 100000).boxed().collect(Collectors.toList()), Types.INT)
                .flatMap((Integer tuple, Collector<Integer> out) -> {
                    IntStream.range(0, 30000).boxed().forEach(num -> out.collect(tuple));
                }).returns(Types.INT)
                .setParallelism(2)
                .print().setParallelism(1);
    }

    private static void backPressureLow(StreamExecutionEnvironment streamEnv) {
        streamEnv.fromCollection(IntStream.range(0, 1000000).boxed().collect(Collectors.toList()), Types.INT)
                .flatMap((Integer tuple, Collector<Integer> out) -> {
                    IntStream.range(0, 3000).boxed().forEach(num -> out.collect(tuple));
                }).returns(Types.INT)
                .setParallelism(2)
                .filter(num -> new Random().nextInt(100) == 30)
                .setParallelism(3)
                .print().setParallelism(1);
    }

    private static void backPressureNo(StreamExecutionEnvironment streamEnv) {
        streamEnv.addSource(new SourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                IntStream.range(0, 1000000).boxed().forEach(num -> {
                    try {
                        Thread.sleep(50);
                        ctx.collect(num);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }

            @Override
            public void cancel() {

            }
        }).flatMap((Integer tuple, Collector<Integer> out) -> {
                    IntStream.range(0, 3000).boxed().forEach(num -> out.collect(tuple));
                    Thread.sleep(50);
                }).returns(Types.INT)
                .setParallelism(2)
                .filter(num -> new Random().nextInt(100) == 30)
                .setParallelism(3)
                .print().setParallelism(1);
    }

}
